package cn.itcast.b_etl.transformation;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

/**
 * 案例:
 * 1.两条流合并 union
 * 2.两条流连接 connect
 */
public class UnionConnectDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /**
         * union,要求多条准备合并的数据流,数据类型必须一致
         */
        DataStreamSource<Integer> s1 = env.fromElements(1, 2, 3, 4, 5, 6);
        DataStreamSource<Integer> s2 = env.fromElements(1, 2, 3, 11, 12, 13, 14, 15, 16);
        DataStreamSource<String> s3 = env.fromElements("a", "a", "a", "a", "a", "a");

//        s1.union(s2).print();
        s1.connect(s3).map(new CoMapFunction<Integer, String, String>() {
            @Override
            public String map1(Integer value) throws Exception {
                return value+"";
            }

            @Override
            public String map2(String value) throws Exception {
                return value;
            }
        }).print();

        env.execute();
    }
}
